package com.j.app.function;/**
 * @ClassName TableProcessFunction
 * @Description TODO
 * @date 2024/3/5 14:24
 * @Author JG
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.j.bean.GmallConfig;
import com.j.bean.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.taskexecutor.JobTable;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * @ClassName TableProcessFunction
 * @Description TODO
 * @date 2024/3/5 14:24
 * @Author JG
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> stringTableProcessMapStateDescriptor;
    private Connection connection;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> stringTableProcessMapStateDescriptor) {
        this.stringTableProcessMapStateDescriptor = stringTableProcessMapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void processBroadcastElement(String s, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
       //1.获取并解析数据
        JSONObject jsonObject = JSON.parseObject(s);
        TableProcess tableprocess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
        checkTable(tableprocess.getSinkTable(),tableprocess.getSinkColumns(),
                   tableprocess.getSinkPk(),tableprocess.getSinkExtend());
        //写入状态并进行广播
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(stringTableProcessMapStateDescriptor);
        broadcastState.put(tableprocess.getSourceTable(),tableprocess);
    }

    /**
     * 检验并且建表
     * @param sinkTable  表名
     * @param sinkColumns  表字段
     * @param sinkPk   表主键
     * @param sinkExtend  表扩展字段
     */
    public void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend){
        PreparedStatement preparedStatement=null;
        try {
            //处理特殊字段
            if(sinkPk==null||"".equals(sinkPk)){
                sinkPk="id";
            }
            if(sinkExtend==null){
                sinkExtend="";
            }
            //拼接sql
            StringBuilder createTableSql = new StringBuilder("create table if not exist ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columns = sinkColumns.split(",");
            for (int i = 0; i < columns.length; i++) {
                String column=columns[i];
                //判断是否为主键
                if(sinkPk.equals(column)){
                    createTableSql.append(column).append(" vachar primary key ");
                }else{
                    createTableSql.append(column).append(" vachar");
                }

                if(i<columns.length-1){
                    createTableSql.append(",");
                }
            }
            createTableSql.append(")").append(sinkExtend);
            //编译sql
            preparedStatement = connection.prepareStatement(createTableSql.toString());


            //执行sql
            preparedStatement.execute();

        } catch (SQLException e) {
            throw new RuntimeException("建表失败"+sinkTable);
        } finally {
            if(preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    throw new RuntimeException(e);
                }
            }
        }

    }

    //value:
    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //1.获取广播流配置数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(stringTableProcessMapStateDescriptor);
        TableProcess tableprocess = broadcastState.get(jsonObject.getString("table"));
        //过滤字段
        if(tableprocess == null) {
            System.out.println("找不到对应的key");
        }else {
            filterColumn(jsonObject.getJSONObject("data"), tableprocess.getSinkColumns());
            jsonObject.put("sinktable",tableprocess.getSinkTable());
            collector.collect(jsonObject);

        }

    }

    //过滤数据
    private void filterColumn(JSONObject data, String sinkColumns) {
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, Object> next = iterator.next();
            if(!sinkColumns.contains(next.getKey())){
                iterator.remove();
            }
        }


    }


    @Override
    public void close() throws Exception {

    }
}
